package net.jueb.frameworkDemo.mq.activeMq.pubSub;

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.TextMessage;

import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.ActiveMQMessageProducer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 *发布者订阅者模式：发布者生产消息放到队列里，多个监听队列的消费者都会收到同一份消息；即正常情况下每个消费者收到的消息应该都是一样的。
 *在默认大多数情况下，AcitveMQ是以异步模式发送消息。例外的情况：在没有使用事务的情况下，生产者以 PERSISTENT传送模式发送消息。在这种情况下，send方法都是同步的，并且一直阻塞直到 ActiveMQ发回确认消息：消息已经存储在持久性数据存储中。这种确认机制保证消息不会丢失，但会造成生产者阻塞从而影响反应时间。 
高性能的程序一般都能容忍在故障情况下丢失少量数据。如果编写这样的程序，可以通过使用异步发送来提高吞吐量（甚至在使用PERSISTENT 传送模式的情况下）。
 * activemq 生产者实例
 */
public class PubClient {
	private static final Logger log = LoggerFactory.getLogger(PubClient.class);
//	private static final String BROKER_URL = ActiveMQConnection.DEFAULT_BROKER_URL;
	private static final String BROKER_URL = "failover:(tcp://192.168.0.205:61616)?maxReconnectAttempts=0&maxReconnectDelay=10000&timeout=-1";
//	private static final String BROKER_URL = "failover:(tcp://192.168.0.205:61616)";
	private static final String TOPIC = "test-activemq-topic";
	
	MessageProducer producer;

	public PubClient(Session session,String topic) throws JMSException {
		// 通过会话创建目标
		Destination dest = session.createTopic(topic);
		// 创建mq消息的消费者
		producer=session.createProducer(dest);
	}
	
	
	public static void main(String[] args) throws JMSException {
		// 初始化连接工厂
		ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(BROKER_URL);
		// 获得连接
//		Connection conn = connectionFactory.createConnection();
		Connection conn = connectionFactory.createConnection("admin", "admin");
		// 启动连接
		conn.start();

		// 创建Session，此方法第一个参数表示会话是否在事务中执行，第二个参数设定会话的应答模式
		Session session = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
		PubClient pubClient=new PubClient(session, TOPIC);
		for (int i = 0; i < 1000; i++) 
		{
			log.debug("send"+i);
			String msg="hello world !"+i;
			// 初始化一个mq消息
			TextMessage message = session.createTextMessage(msg);
			// 发送消息
			try {
				pubClient.send(message,1000);
			} catch (Exception e) {
				log.error(e.getMessage(),e);
			}
			log.debug("send:"+msg);
			try {
				Thread.sleep(1000);
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
		log.debug("close");
		// 关闭mq连接
		conn.close();
	}
	
	public void send(Message msg) throws JMSException
	{
		producer.send(msg);
	}
	
	public void send(Message msg,int i,int x,int y) throws JMSException
	{
		producer.send(msg,i,x,y);
	}
	
	public void send(Message msg,int timeOut) throws JMSException
	{
		ActiveMQMessageProducer aProducer=(ActiveMQMessageProducer)producer;
		int oldTimeOut=aProducer.getSendTimeout();
		aProducer.setSendTimeout(timeOut);
		producer.send(msg);
		aProducer.setSendTimeout(oldTimeOut);
	}
}
